//
// Created by dave on 18-8-7.
//
#pragma once
#include <atomic>
#include <memory>
#include <string>
#include <vector>
#include <mutex>
#include <boost/any.hpp>
#include <glog/logging.h>
#include <muduo/net/EventLoop.h>
#include <muduo/net/TcpServer.h>
#include "jtt808_server.h"
#include "codec/data_pack.h"
#include "codec/data_utils.h"
#include "codec/platform/plat_common_resp.h"
#include "codec/terminal/location_report.h"
#include "codec/terminal/term_register.h"
#include "codec/platform/term_register_resp.h"

using namespace muduo;
using namespace muduo::net;

class Jtt808Server {
public:
    Jtt808Server(EventLoop* loop,
                 const InetAddress& listenAddr,
                 uint32_t numThreads) : server_(loop, listenAddr, "jtt808Server") {
        server_.setThreadNum(numThreads);
        server_.setConnectionCallback(std::bind(&Jtt808Server::onConnection, this, std::placeholders::_1));
        server_.setMessageCallback(std::bind(&Jtt808Server::onMessage, this,
                std::placeholders::_1, std::placeholders::_2, std::placeholders::_3));
        server_.setThreadInitCallback(std::bind(&Jtt808Server::threadInit, this, std::placeholders::_1));
    }

    void start() {
        server_.start();
    }

private:
    void onConnection(const TcpConnectionPtr& conn) {
        if (conn->connected()) {
            conn->setTcpNoDelay(true);
            conn->setContext(boost::any(0));
        }
    }

    void onMessage(const TcpConnectionPtr& conn, Buffer* buf, Timestamp ts) {
        size_t size = buf->readableBytes();
        if (size > 4 * 1024 * 1024) {
            buf->retrieveAll();
            return;
        }

        const char* p = buf->peek();
        size_t start = _find_mark(p, 0, size);
        if (start == 0 && p[0] != 0x7E) {
            buf->retrieveAll();
        } else {
            do {
                size_t end = _find_mark(p, start + 1, size);
                if (end > 0) {
                    if (end - start == 1) {
                        start = end;
                        continue;
                    }

                    onFrame(conn, p + start, end - start + 1);
                    start = _find_mark(p, end + 1, size);
                    if (start == 0) {
                        buf->retrieve(end + 1);
                        break;
                    }
                } else {
                    buf->retrieve(start);
                    break;
                }
            } while (true);
        }
    }

    void onFrame(const TcpConnectionPtr& conn, const char* ptr, size_t length) {
        int count = ++msgCounter_;
        if (count % 10000 == 0) {
            LOG(INFO) << "接收(万包)：" << count / 10000;
        }

        if (length < 14) {
            LOG(WARNING) << "无效桢：" << length << "字节";
            return;
        }

        DataPack dp = decode_data_pack(ptr, 0, length);
        auto seq = boost::any_cast<int*>(conn->getMutableContext());
        if (*seq >= 0xFFFF) {
            *seq = 0;
        }
        if (dp.getMsgId() == 0x0200) {
            auto lr = std::dynamic_pointer_cast<LocationReport>(dp.getBody());
            // LOG(INFO) << dp.getSeqNo() << "|" << dp.getSim() << ": " << lr->getLng() << ", " << lr->getLat();
            // location loc(dp);
    //        if (!saver_.save(loc)) {
    //            LOG(WARNING) << "丢弃1条";
    //        }

            DataPack re;
            re.setMsgId(0x8001);
            re.setSeqNo(static_cast<unsigned short>(++*seq));
            re.setSim(dp.getSim());

            auto body = std::make_shared<PlatCommonResp>();
            body->setTermMsgId(dp.getMsgId());
            body->setTermSeqNo(dp.getSeqNo());
            body->setResult(0);
            re.setBody(body);

            std::vector<char> vec = re.encode();
            conn->send(vec.data(), static_cast<int>(vec.size()));
        } else if (dp.getMsgId() == 0x0100) {
            std::shared_ptr<TermRegister> tr = std::dynamic_pointer_cast<TermRegister>(dp.getBody());
            // LOG(INFO) << dp.getSeqNo() << "|" << dp.getSim() << ": " << tr->getLicense();

            DataPack re;
            re.setMsgId(0x8100);
            re.setSeqNo(static_cast<unsigned short>(++*seq));
            re.setSim(dp.getSim());

            auto resp = std::make_shared<TermRegisterResp>();
            resp->setTermSeqNo(dp.getSeqNo());
            resp->setResult(0);
            resp->setAuth("abcdef");
            re.setBody(resp);

            std::vector<char> vec = re.encode();
            conn->send(vec.data(), static_cast<int>(vec.size()));
        } else {
            LOG(INFO) << "ERROR: " << dp.getMsgId();
        }
    }


    void threadInit(EventLoop* loop) {
        std::unique_lock<std::mutex> lock(mutex_);
        loops_.push_back(loop);
    }

    size_t _find_mark(const char* ptr, size_t start, size_t stop) {
        for (size_t i = start; i < stop; i++) {
            if (ptr[i] == 0x7E) {
                return i;
            }
        }
        return 0;
    }

private:
    std::atomic_int msgCounter_{0};
    TcpServer server_;
    std::mutex mutex_;
    std::vector<EventLoop*> loops_;
};


